package com.my.func;

import com.alibaba.fastjson.JSONObject;
import org.apache.commons.math3.fitting.leastsquares.EvaluationRmsChecker;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class WithStateKeyedProcessFunction extends KeyedProcessFunction<String, JSONObject,JSONObject> {

    private ValueState<String> stringValueState;
    private ValueState<Long> longValueState;
    private ValueState<Boolean> hasInit;
    private MapState<String,String> mapState;

    public WithStateKeyedProcessFunction() {
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        //为state设置生命周期，清除过期数据
        StateTtlConfig ttl = StateTtlConfig.newBuilder(Time.hours(24L)).setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite).build();

        ValueStateDescriptor<String> stringValueStateDescriptor = new ValueStateDescriptor<>("string-value-state", Types.STRING);
        stringValueStateDescriptor.enableTimeToLive(ttl);
        stringValueState=getRuntimeContext().getState(stringValueStateDescriptor);


    }

    @Override
    public void processElement(JSONObject jsonObject, Context context, Collector<JSONObject> collector) throws Exception {
        collector.collect(jsonObject);
    }
}
